/**
 * 
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2014-11-4]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
package com.ailk.bigdata.etl.realstream.server.model;

import java.io.Serializable;
import java.lang.reflect.Array;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2015-7-9]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class RingBufferQueue<T> implements Serializable
{
	private static final long serialVersionUID = -4842205158902601162L;

	private volatile AtomicInteger head;

	private volatile AtomicInteger tail;

	private int length;

	final T EMPTY = null;

	private volatile T[] queue;

	@SuppressWarnings("unchecked")
	public RingBufferQueue(Class<T> type, int length)
	{
		this.head = new AtomicInteger(0);
		this.tail = new AtomicInteger(0);
		this.length = length == 0 ? 2 << 16 : length; // 默认2^16
		this.queue = (T[]) Array.newInstance(type, this.length);
	}

	@SuppressWarnings("unchecked")
	public void enQueue(T t)
	{
		if (t == null)
			t = (T) new Object();
		// 阻塞 -- 避免多生成者循环生产同一个节点
		while (this.getTail() - this.getHead() >= this.length)
			;
		int ctail = this.tail.getAndIncrement();
		while (this.queue[this.getTail(ctail)] != EMPTY)
			;
		this.queue[this.getTail(ctail)] = t;
	}

	public T deQueue()
	{
		T t = null;
		// 阻塞 -- 避免多消费者循环消费同一个节点
		while (this.head.get() >= this.tail.get())
		{
			if (this.head.get() > 0 && this.tail.get() < 0)
			{
				break;
			}
		}
		;
		int chead = this.head.getAndIncrement();
		while (this.queue[this.getHead(chead)] == EMPTY)
			;
		t = this.queue[this.getHead(chead)];
		this.queue[this.getHead(chead)] = EMPTY;
		return t;
	}

	public int getHead(int index)
	{
		return index & (this.length - 1);
	}

	public int getTail(int index)
	{
		return index & (this.length - 1);
	}

	public int getHead()
	{
		return head.get() & (this.length - 1);
	}

	public int getTail()
	{
		return tail.get() & (this.length - 1);
	}

	public T[] getQueue()
	{
		return queue;
	}

	public int getLength()
	{
		return length;
	}

	public void setLength(int length)
	{
		this.length = length;
	}
}
